package cn.itcast.tags.lu

import cn.itcast.tags.models.{AbstractModel, ModelType}
import org.apache.spark.ml.clustering.{KMeans, KMeansModel}
import org.apache.spark.ml.feature.VectorAssembler
import org.apache.spark.sql.{Column, DataFrame, SparkSession}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.DataTypes

/**
  * 客户价值模型标签 RFM
  * 370	客户价值
  * 371	高价值		0
  * 372	中上价值		1
  * 373	中价值		2
  * 374	中下价值		3
  * 375	低价值		4
  */
class RfmModel extends AbstractModel("客户价值模型标签", ModelType.ML) {
  override def doTag(businessDF: DataFrame, tagDF: DataFrame): DataFrame = {
    import businessDF.sparkSession.implicits._

    //businessDF.printSchema()
    /**
      * root
      * -- memberid: string (nullable = true)
      * -- ordersn: string (nullable = true)
      * -- orderamount: string (nullable = true)
      * -- finishtime: string (nullable = true)
      */
    //businessDF.show()
    /**
      * +--------+--------------------+-----------+----------+
      * |memberid|             ordersn|orderamount|finishtime|
      * +--------+--------------------+-----------+----------+
      * |     365|gome_792756751164275|    2479.45|1590249600|
      * |     783|jd_14090106121770839|    2449.00|1591545600|
      * |     562|jd_14090112394810659|    1099.42|1590595200|
      */
    // todo 1、解析业务数据，得到RFM(Recency Frequency Monetary)
    val rfmDF: DataFrame = businessDF
      //按照用户id分组
      .groupBy($"memberid")
      //计算RFM值
      .agg(
      //R值
      max($"finishtime").as("finish_time"),
      //F值
      count($"ordersn").as("frequency"),
      //M值
        sum($"orderamount".cast(DataTypes.createDecimalType(10, 2))).as("monetary")
    )
      .select(
        $"memberid".as("uid"),
        datediff(current_timestamp(), from_unixtime($"finish_time")).as("recency"),
        $"frequency", $"monetary"
      )
    // rfmDF.printSchema()
    /**
      * root
      * |-- memberid: string (nullable = true)
      * |-- recency: integer (nullable = true)
      * |-- frequency: long (nullable = false)
      * |-- monetary: decimal(10,2) (nullable = true)
      */
    // rfmDF.show(20,false)
    /**
      * +--------+-------+---------+---------+
      * |memberid|recency|frequency|monetary |
      * +--------+-------+---------+---------+
      * |1       |77     |226      |524701.09|
      * |102     |77     |113      |153706.15|
      * |107     |77     |108      |209866.94|
      */
    // todo 2、对RFM根据规则进行打分(features数值化)
    /**
      * R: 1-3天=5分，4-6天=4分，7-9天=3分，10-15天=2分，大于16天=1分
      * F: ≥200=5分，150-199=4分，100-149=3分，50-99=2分，1-49=1分
      * M: ≥20w=5分，10-19w=4分，5-9w=3分，1-4w=2分，<1w=1分
      */
    //打分表达式
    val r: Column = when($"recency".between(1, 5), 5.0)
      .when($"recency".between(4, 6), 4.0)
      .when($"recency".between(7, 9), 3.0)
      .when($"recency".between(10, 15), 2.0)
      .when($"recency".gt(16), 1.0)
    val f: Column = when($"frequency".geq(200), 5.0)
      .when($"frequency".between(150, 199), 4.0)
      .when($"frequency".between(100, 149), 3.0)
      .when($"frequency".between(50, 99), 2.0)
      .when($"frequency".between(1, 49), 1.0)
    val m = when($"monetary".geq(200000), 5.0)
      .when($"monetary".between(100000, 199999), 4.0)
      .when($"monetary".between(50000, 99999), 3.0)
      .when($"monetary".between(10000, 49999), 2.0)
      .when($"monetary".lt(10000), 1.0)
    // 对Recency Frequency Monetary 进行打分
    val rfmScoreDF: DataFrame = rfmDF.select(
      $"uid",
      r.as("r_score"),
      f.as("f_score"),
      m.as("m_score")
    )

    // rfmScoreDF.printSchema()

    /**
      * root
      * |-- uid: string (nullable = true)
      * |-- r_score: double (nullable = true)
      * |-- f_score: double (nullable = true)
      * |-- m_score: double (nullable = true)
      */
    // rfmScoreDF.show(20,false)
    /**
      * +---+-------+-------+-------+
      * |uid|r_score|f_score|m_score|
      * +---+-------+-------+-------+
      * |1  |1.0    |5.0    |5.0    |
      * |102|1.0    |3.0    |4.0    |
      * |107|1.0    |3.0    |5.0    |
      * |110|1.0    |3.0    |5.0    |
      */
    // todo 3、特征向量化 使用KMeans算法进行构建模型

    val assembler = new VectorAssembler()
      .setInputCols(Array("r_score", "f_score", "m_score"))
      .setOutputCol("features")

    val featuresDF = assembler.transform(rfmScoreDF)



    //featureDF.printSchema()
    /**
      * root
      * |-- uid: string (nullable = true)
      * |-- r_score: double (nullable = true)
      * |-- f_score: double (nullable = true)
      * |-- m_score: double (nullable = true)
      * |-- feature: vector (nullable = true)
      */
    //featureDF.show(20,false)
    /**
      * +---+-------+-------+-------+-------------+
      * |uid|r_score|f_score|m_score|feature      |
      * +---+-------+-------+-------+-------------+
      * |1  |1.0    |5.0    |5.0    |[1.0,5.0,5.0]|
      * |102|1.0    |3.0    |4.0    |[1.0,3.0,4.0]|
      * |107|1.0    |3.0    |5.0    |[1.0,3.0,5.0]|
      */
    val kmeanModel: KMeansModel =new KMeans()
      .setFeaturesCol("features") //对那一列进行模型训练
      .setPredictionCol("prediction") //模型训练结果预测列
      .setK(5) //KMeans的K
      .setMaxIter(20) //设置最大迭代次数
      .setInitMode("k-means||")
      .fit(featuresDF)
    val wssse: Double = kmeanModel.computeCost(featuresDF)
    println(s"WSSSE = $wssse")

    val model: DataFrame = kmeanModel.transform(featuresDF)
    model.printSchema()
    model.show(10,false)

    null
  }
}

object RfmModel {
  def main(args: Array[String]): Unit = {
    val model = new RfmModel()
    model.executeModel(355L)
  }
}
